package xin.alum.aim.rocketmq;

import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
import xin.alum.aim.Aim;
import xin.alum.aim.groups.Session;
import xin.alum.aim.groups.SessionGroups;
import xin.alum.aim.groups.Sessions;
import xin.alum.aim.model.Transportable;
import java.nio.charset.StandardCharsets;
import java.util.Collections;
import java.util.Objects;


@Slf4j
public class RocketMqConsumer  {

   private final Sessions sessions;
    private final SessionGroups groups;
    private final Jackson2JsonRedisSerializer<Object> jackson;
    private final ClientServiceProvider provider;
    private final RocketMQConfig config ;
    private final ClientConfiguration configuration;

    public RocketMqConsumer(RocketMQConfig config,ClientServiceProvider provider,ClientConfiguration configuration,Jackson2JsonRedisSerializer<Object> json,Sessions sessions,SessionGroups groups) {
        this.config=config;
        this.provider=provider;
        this.configuration=configuration;
        this.sessions=sessions;
        this.groups=groups;
        this.jackson=json;
        Consumer("*");
    }

    public void Consumer(String tag){
        try {
            // 订阅消息的过滤规则，表示订阅所有Tag的消息。
            FilterExpression filterExpression = new FilterExpression(tag, FilterExpressionType.TAG);
            provider.newPushConsumerBuilder()
                    .setClientConfiguration(configuration)
                    // 设置消费者分组。
                    .setConsumerGroup(config.consumerGroup)
                    // 设置预绑定的订阅关系。
                    .setSubscriptionExpressions(Collections.singletonMap(config.topic, filterExpression))
                    // 设置消费监听器。
                    .setMessageListener(message -> {
                        try {
                            if(log.isDebugEnabled()) {
                                log.info("消费消息:" + StandardCharsets.UTF_8.decode(message.getBody()));
                            }
                            byte[] arr = new byte[message.getBody().remaining()];
                            message.getBody().get(arr);
                            if(message.getTag().isPresent()&&message.getTag().get().equals(RocketMQConfig.TAG_KILL))
                            {
                                sessions.kick((Session) Objects.requireNonNull(jackson.deserialize(arr)));
                            }else {
                                Transportable t = (Transportable) jackson.deserialize(arr);
                                if (Aim.message != null) {
                                    Aim.message.onMessage(arr);
                                } else if (t != null) {
                                    if (t.getRecipient().isEmpty() || t.getRecipient().equals(Sessions.ALL_SESSIONS)) {
                                        sessions.sends(t);
                                    } else {
                                        groups.get(t.getRecipient()).sends(t);
                                    }
                                }
                            }
                        } catch (Exception e) {
                            log.error("消费异常：{0}",e);
                        }

                        return ConsumeResult.SUCCESS;
                    }).build();
            log.info("构建MQ5.0消费者成功：proxy:{}, topic:{}, group：{}", config.proxy, config.topic, config.consumerGroup);
        } catch (ClientException e) {
            log.error("构建MQ5.0消费者异常：proxy:{}, topic:{}, group：{}", config.proxy, config.topic, config.consumerGroup, e);
        }
    }
}